#include "excutor.hpp"

#include <unordered_map>
#include <stdexcept>
#include <thread>
#include <atomic>
#include <mutex>
#include <unistd.h>
#include <sys/epoll.h>
#include <fcntl.h>

#include "loger.hpp"

class LinuxExcutor : public sl::Excutor
{
private:
    int fd_ = -1;
    int pipe_fd_[2] = {-1, -1}; // 管道：[0]读端，[1]写端

    std::unordered_map<int, Callback> callback_;

    std::atomic<bool> running_ = true;
    std::thread thread_;
    std::mutex mutex_;

public:
    LinuxExcutor()
    {   
        fd_ = epoll_create1(EPOLL_CLOEXEC);
        if(fd_ < 0) {
            log_error("create epoll failed");
            throw std::system_error(errno, std::system_category(), "epoll create failed");
        }

        if(pipe2(pipe_fd_, O_CLOEXEC | O_NONBLOCK) == -1) {
            log_error("create pipe failed");
            throw std::system_error(errno, std::system_category(), "epoll pipe failed");
        }

        epoll_add(pipe_fd_[0], EPOLLIN | EPOLLET | EPOLLONESHOT);

        thread_ = std::thread(&LinuxExcutor::worker, this, -1);
    }
    ~LinuxExcutor()
    {
        if(running_.exchange(false)) {
            char dummy[1] = {'x'};
            ::write(pipe_fd_[1], dummy, 1);

            if(thread_.joinable()) {
                thread_.join();
            }
        }

        if(fd_ > 0) {
            ::close(fd_);
        }
        if(pipe_fd_[0] > 0) {
            ::close(pipe_fd_[0]);
        }
        if(pipe_fd_[0] > 0) {
            ::close(pipe_fd_[1]);
        }
    }

public:
    bool is_running() const override {
        return running_;
    }

    void submit(std::any handle, Callback cb) override
    {
        int fd = std::any_cast<int>(handle);
        if(fd < 0) {
            log_error("invalid handle");
            return;
        }
        std::lock_guard<std::mutex> lock(mutex_);
        callback_[fd] = std::move(cb);
        epoll_add(fd, EPOLLET | EPOLLIN);
    }

    void remove(std::any handle) override {
        int fd = std::any_cast<int>(handle);
        std::lock_guard<std::mutex> lock(mutex_);

        epoll_remove(fd);
        callback_.erase(fd);
    }

    private:
        void epoll_add(int fd, int event) {
            struct epoll_event ev {};
            ev.data.fd = fd;
            ev.events = event;
            // ev.events |= EPOLLONESHOT;
            if (epoll_ctl(fd_, EPOLL_CTL_ADD, fd, &ev) == -1) {
                if (errno == EEXIST) {
                    if (epoll_ctl(fd_, EPOLL_CTL_MOD, fd, &ev) == -1) {
                        log_error("epoll_ctl MOD failed");
                    }
                } else {
                    log_error("epoll_ctl ADD failed");
                }
            }
        }
        
        void epoll_remove(int fd) {
            if (epoll_ctl(fd_, EPOLL_CTL_DEL, fd, nullptr) == -1) {
                if (errno == ENOENT) {
                    // 忽略 "不存在" 错误 ?
                    return;
                } else {
                    log_error("epoll_ctl DEL failed");
                    return;
                }
            }
        }
        
        void epoll_mod(int fd, int event) {
            struct epoll_event ev {};
            ev.data.fd = fd;
            ev.events = event;
            if (epoll_ctl(fd_, EPOLL_CTL_MOD, fd, &ev) == -1) {
                log_error("epoll_ctl MOD failed");
            }
        }

        void set_nonblocking(int fd) {
            if (fd < 0) {
                return;
            }

            int flags = fcntl(fd, F_GETFL, 0);
            if (flags == -1) {
                return;
            }

            if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
                return;
            }
        }

        int worker(int timeout_ms = -1)
        {
            constexpr int max_events = 128;
            std::vector<struct epoll_event> events(max_events);

            while (running_) {
                int nfds = epoll_wait(fd_, events.data(), max_events, timeout_ms);
                if (nfds == -1) {
                    if (errno == EINTR) {
                        log_error("epoll wait interupt by signal");
                        return 0;
                    }
                    log_error("epoll wait failed");
                    return ERR_LINE;
                }

                // 处理触发的事件
                for (int i = 0; i < nfds; ++i) {
                    int fd = events[i].data.fd;

                    // 查找回调函数
                    std::lock_guard<std::mutex> lock(mutex_);
                    auto it = callback_.find(fd);
                    if (it != callback_.end()) {
                        try {
                            it->second();
                        } catch (const std::exception& e) {
                            log_error("callback for {} threw exception:{}", fd, e.what());
                        } catch (...) {
                            log_error("callback for {} threw unknown exception", fd);
                        }
                    }
                }
            }

            log_info("epoll loop exit");
            return 0;
        }

        void pipe_handle() {
            char buf[8];
            while (::read(pipe_fd_[0], buf, sizeof(buf)) > 0)
            {
            }
            
            if(!running_) {
                log_debug("recv pipe signal");
            }
        }
    };

const sl::Excutor& sl::excutor(){
    static LinuxExcutor ex;
    return ex;
}
